package com.github.llyb120.yaomq.server;

import cn.hutool.core.io.IoUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONObject;
import com.alibaba.fastjson.JSON;
import lombok.Cleanup;
import lombok.SneakyThrows;
import lombok.Synchronized;
import org.springframework.stereotype.Service;
import sun.nio.ch.IOUtil;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import javax.websocket.Session;
import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

/**
 * @Author: Administrator
 * @Date: 2020/9/14 11:51
 */
public class MQService {

    public static final String URL_CONNECT = "_connect";
    public static final String KEY_PROXY_NAME = "yaomq-name";
    public static final String KEY_PROXY_TIMEOUT = "yaomq-timeout";
    public static final String KEY_PROXY_ALL = "yaomq-broadcast";

    public static final int CODE_ALL_SUCCESS = 2;
    public static final int CODE_ONE_SUCCESS = 1;
    public static final int CODE_FAILED = -1;

    public static ExecutorService executorService = Executors.newCachedThreadPool();
    public static ReentrantLock lock = new ReentrantLock();

    private static Map<String, Set<Session>> sessionMap = new HashMap<>();
    public static Map<String, MessageBuffer> messageBufferMap = new ConcurrentHashMap<>();

    @SneakyThrows
    public static int proxyWithSocket(HttpServletRequest request, HttpServletResponse response) {
        try{
            //name
            String clientName = request.getHeader(KEY_PROXY_NAME);
//            if (StrUtil.isEmpty(clientName)) {
//                return CODE_FAILED;
//            }
            //timeout
            int timeout = 30;
            String strTimeout = request.getHeader(KEY_PROXY_TIMEOUT);
            if (!StrUtil.isEmpty(strTimeout)) {
                timeout = NumberUtil.parseInt(strTimeout);
            }
            //sendall
            boolean sendAll = false;
            String _sendAll = request.getHeader(KEY_PROXY_ALL);
            if (!StrUtil.isEmpty(_sendAll)) {
                sendAll = true;
            }
            boolean noName = StrUtil.isEmpty(clientName);
            if(!sendAll && noName){
                return CODE_FAILED;
            }
            Map<String, String> headers = new HashMap<>();
            Enumeration<String> it = request.getHeaderNames();
            while(it.hasMoreElements()){
                String key = it.nextElement();
                if(key.startsWith("yaomq-")){
                    continue;
                }
                headers.put(key, request.getHeader(key));
            }
            MessageBuffer messageBuffer = new MessageBuffer();
            messageBuffer.setMethod(request.getMethod());
            messageBuffer.setUrl(request.getRequestURI() + (null == request.getQueryString() ? "" : "?" + request.getQueryString()));
            @Cleanup InputStream is = request.getInputStream();
            messageBuffer.setBody(IoUtil.read(is, StandardCharsets.ISO_8859_1));
            //bind ACK and thread
            String ack = IdUtil.objectId();
            messageBuffer.setAck(ack);
            messageBuffer.setExpr(timeout * 1000 + System.currentTimeMillis() + 5000);
            String json = JSON.toJSONString(messageBuffer);
            //如果是发送给所有人，那么不需要返回值
            if(sendAll){
                if(noName){
                    for (Session session : getSessionSet()) {
                        session.getBasicRemote().sendText(json);
                    }
                } else {
                    for (Session session : getSessionSet(clientName)) {
                        session.getBasicRemote().sendText(json);
                    }
                }
                return CODE_ALL_SUCCESS;
            }
            messageBufferMap.put(ack, messageBuffer);
            lock.lock();
            Condition condition = lock.newCondition();
            messageBuffer.setCondition(condition);
            Session session = getSession(clientName);
            session.getBasicRemote().sendText(json);
            condition.await(timeout, TimeUnit.SECONDS);
            if(!messageBuffer.isResponse()){
                return CODE_FAILED;
            }
            //copy status
            response.setStatus(messageBuffer.getStatus());
            //copy header
            messageBuffer.getHeaders().forEach((k,v) -> {
                response.setHeader(k,v);
            });
            //copy body
            @Cleanup OutputStream os = response.getOutputStream();
            IoUtil.copy(new StringBufferInputStream(messageBuffer.getBody()), os);

            return CODE_ONE_SUCCESS;
        } finally {
            lock.unlock();
        }

    }

    public static void handleServerMessage(Session session, String message){
        MessageBuffer revBuffer = JSON.parseObject(message, MessageBuffer.class);
        MessageBuffer oldBuffer = messageBufferMap.remove(revBuffer.getAck());
        if (oldBuffer == null) {
            return;
        }
        oldBuffer.setBody(revBuffer.getBody());
        oldBuffer.setHeaders(revBuffer.getHeaders());
        oldBuffer.setStatus(revBuffer.getStatus());
        oldBuffer.setResponse(true);
        lock.lock();
        oldBuffer.getCondition().signalAll();
        lock.unlock();
    }

    @Synchronized
    public static void saveSession(String name, Session session){
        if (name == null) {
            return;
        }
        Set<Session> list = sessionMap.computeIfAbsent(name, k -> new HashSet<>());
        list.add(session);
    }

    @Synchronized
    public static void cleanSession(Session session){
        sessionMap.forEach((k,v) -> {
            v.remove(session);
        });
    }

    @Synchronized
    public static Set<Session> getSessionSet(){
        return sessionMap.values()
            .stream()
            .flatMap(Collection::stream)
            .collect(Collectors.toSet());
    }

    @Synchronized
    public static Set<Session> getSessionSet(String name){
        if (name == null) {
            return new HashSet<>();
        }
        Set<Session> set = sessionMap.get(name);
        if (set == null) {
            return new HashSet<>();
        }
        return new HashSet<>(set);
    }

    @Synchronized
    public static Session getSession(String name){
        Set<Session> set = getSessionSet(name);
        int num = RandomUtil.randomInt(0, set.size());;
        for (Session session : set) {
            if(num-- > 0){
                continue;
            }
            return session;
        }
        return null;
    }
}
